package com.lizhiyu.flink.demo4_transformation;

import com.lizhiyu.flink.demo2_source.CustomSource;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import scala.Tuple2;
import scala.Tuple3;

import java.util.Date;


/***
 * 使用 flapMap 和 map 进行转换数据
 * 1、flatMap 使用 RichFlatMapFunction,map 使用 RichMapFunction
 * 2、带Rich的接口扩展更多
 */
public class FaltMapTransformMation {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> ds = env.addSource(new CustomSource());

        SingleOutputStreamOperator singleOutputStreamOperator = ds.flatMap(new RichFlatMapFunction<String, Tuple2<String,String>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, String>> out) throws Exception {
                String[] split = value.split(":");
                Tuple2 tuple2 = new Tuple2(split[0],split[1]);
                out.collect(tuple2);
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("start");
            }

            @Override
            public void close() throws Exception {
                System.out.println("end");
            }
        }).setParallelism(1);

        singleOutputStreamOperator.print();


        SingleOutputStreamOperator mapDS = ds.map(new RichMapFunction<String,Tuple2<String,String>>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("map open");
            }

            @Override
            public void close() throws Exception {
                System.out.println("flapmap close");
            }
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                String[] split = value.split(":");
                Tuple2 tuple2 = new Tuple2(split[0],split[1]);
                return tuple2;
            }
        });

        mapDS.print();
        env.execute("test");
    }
}
